package cn.com.bonc.khjy.controller;


import cn.com.bonc.khjy.bean.RestFullBean;
import cn.com.bonc.khjy.config.KafkaConfig;
import cn.com.bonc.khjy.schedu.FileFTPMonitor;
import cn.com.bonc.khjy.schedu.RestFullMonitor;
import cn.com.bonc.khjy.utils.JacksonMapper;
import cn.com.bonc.khjy.utils.PrintUtil;
import kafka.admin.AdminUtils;
import kafka.utils.ZkUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.security.JaasUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import org.springframework.stereotype.Component;

import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.*;

@Component
public class KafkaController {

    //重试间隔
    private final static Long TIMEOUT_SECOND = 120000L;
    private static Logger logger = LoggerFactory.getLogger( KafkaController.class );
    @Value( "${zookeeper.connect}" )
    private String zkServers;
    private ZkUtils zkUtils;
    @Autowired
    private KafkaConfig config;
   /* @Autowired
    private CommonService service;*/
    @Autowired
    private FileFTPMonitor fileFTPMonitor;
    @Autowired
    private RestFullMonitor restFullMonitor;


    /**
     * @Desc:判断某个Topic是否存在
     * @Param: topicName
     * @Return: boolean exist
     */

    public boolean topicExists( String topicName ) {
        zkUtils = ZkUtils.apply( zkServers, 50000, 50000, JaasUtils.isZkSecurityEnabled() );
        boolean exists = AdminUtils.topicExists( zkUtils, topicName );
        return exists;
    }

    /**
     * @param topics
     * @Desc: 订阅一批topic并消费数据
     */
    public void getData( List< String > topics ) {

        Long currentTimeMillis = System.currentTimeMillis();
        logger.debug( "FTP-订阅的Topics为:{}", topics );
        KafkaConsumer< String, String > consumer = config.consumerFactory();
        consumer.subscribe( topics );
        long start = System.currentTimeMillis();
        logger.debug( "FTP-开始从kafka消费数据，开始时间: {}", System.currentTimeMillis() );
        try{
            while ( true ) {
                ConsumerRecords< String, String > records = consumer.poll( Duration.ofMillis( 10 ) );
                for ( ConsumerRecord< String, String > record : records ) {

                    //logger.debug( "FTP-KAFKA消费数据 ->>> topic= {}数据:{},{}", record.topic(), record.offset(), record.value() );
                    String str = parseJsonToFTPString(record.value());
                    //logger.debug("写到FTP的数据:{}",str);
                    fileFTPMonitor.addQueueLine( str );
                    consumer.commitSync();
                }
            }
        }catch (Exception ex){
            logger.error("kafka消费数据", ex);
        }finally {
            consumer.commitSync();
            consumer.close();
        }
        logger.debug( "结束从kafka消费数据，消耗时间: {}", System.currentTimeMillis() - start );
        logger.debug( "本次任务执行完成，等待下一次任务触发。" );
    }

    public void getDataRest(){
        List<RestFullBean> ls=restFullMonitor.getRestRuleList();
        List< String > topics=new ArrayList<>();
        for (RestFullBean restFullBean:
        ls) {
            topics.add(restFullBean.getTopic());
        }
        Long currentTimeMillis = System.currentTimeMillis();
        logger.debug( "getDataRest-订阅的Topics为:{}", topics );
        KafkaConsumer< String, String > consumer = config.consumerFactory();
        consumer.subscribe( topics );
        long start = System.currentTimeMillis();
        logger.debug( "getDataRest-开始从kafka消费数据，开始时间: {}", System.currentTimeMillis() );
        while ( true ) {
             if ( ( currentTimeMillis + TIMEOUT_SECOND ) > System.currentTimeMillis() ) {
                ConsumerRecords< String, String > records = consumer.poll( Duration.ofMillis( 10 ) );
                for ( ConsumerRecord< String, String > record : records ) {
                    currentTimeMillis = System.currentTimeMillis();
                    logger.debug( "getDataRest-KAFKA消费数据 ->>> topic= {}数据:{},{}", record.topic(), record.offset(), record.value() );
                    String str = parseJsonToRSString(record.value());
                    restFullMonitor.addQueueLine( str );
                }
            } else {
                break;
            }
        }
        consumer.close();
        logger.debug( "getDataRest-结束从kafka消费数据，消耗时间: {}", System.currentTimeMillis() - start );
        logger.debug( "getDataRest-本次任务执行完成，等待下一次任务触发。" );
    }

    public       String parseJsonToRSString(String jsonData){
        Map<String,Object> bodyPar=new HashMap<>();
        Map<String,Object> bodyPar1=new HashMap<>();
        Map<String, ?> map = JacksonMapper.INSTANCE.readJsonToObject(jsonData);
        String str = String.valueOf(map.get("data"));
        Map< String, ? > data = JacksonMapper.INSTANCE.readJsonToObject(str);
        List<RestFullBean> ls=restFullMonitor.getRestRuleList();
         for (RestFullBean restFullBean:
                ls) {
           String cid=restFullBean.getId();
            if (!StringUtils.isEmpty(map.get( "channelCode" ))){
                String id =  map.get( "channelCode" ).toString();
                 if(cid.equals(id)){
                     List<Map> pars= restFullBean.getBodyParameter();
                     for(int i=0;i<pars.size();i++){
                         Map temp=pars.get(i);
                         String key=String.valueOf(temp.get("key"));
                         if(key.equals("data")){

                             String[] args=String.valueOf(temp.get("value")).split(",");
                             for(int j=0;j<args.length;j++){
                                 if(PrintUtil.isNotBlank(String.valueOf(data.get(args[j])))){
                                     Map<String,String> content=new HashMap<>();
                                     content.put("value",String.valueOf(data.get(args[j])));
                                     bodyPar.put(args[j],content);
                                 }
                                 if (!StringUtils.isEmpty(map.get(args[j] ))){   // 话术
                                     Map<String,String> content=new HashMap<>();
                                     content.put("value",String.valueOf(map.get(args[j] )));
                                     bodyPar.put(args[j],content);
                                 }

                             }

                         }else{
                             bodyPar1.put(key,String.valueOf(temp.get("value")));
                         }

                     }
                     bodyPar1.put("data",bodyPar);
                     bodyPar1.put("id",id);
                 }
            }
        }





        return JacksonMapper.INSTANCE.writeObjectToJson(bodyPar1);
    }

    public String parseJsonToFTPString(String jsonData){

      /*  Map<String, ?> map = JacksonMapper.INSTANCE.readJsonToObject(jsonData);
        String str = String.valueOf(map.get("data"));
        Map< String, ? > data = JacksonMapper.INSTANCE.readJsonToObject(str);
        StringBuilder sb = new StringBuilder();
        String activityRemarks =  String.valueOf(map.get( "activityRemarks" ));
        if (!StringUtils.isEmpty(map.get( "channelCode" ))){
            String id =  map.get( "channelCode" ).toString();
             sb.append(id);
        }
        if(PrintUtil.isNotBlank(activityRemarks)){

            sb.append("|").append(activityRemarks);
        }else{
            logger.debug("activityRemarks 关键参数为空");
            sb.append("|").append("RG");
        }

        sb.append("|").append(PrintUtil.isNotBlank(String.valueOf(data.get("对象类型")))?String.valueOf(data.get("对象类型")):" ");
        sb.append("|").append(PrintUtil.isNotBlank(String.valueOf(data.get("关联主卡号码")))?String.valueOf(data.get("关联主卡号码")):" ");
        sb.append("|").append("755");
        sb.append("|").append(PrintUtil.isNotBlank(String.valueOf(data.get("执行人工号")))?String.valueOf(data.get("执行人工号")):" ");
*/
        return jsonData;
    }

    /**
     *
     * @Desc 数据入库
     */

  /*  private void parseJson( String jsonData ) {

        Map< String, ? > map = JacksonMapper.INSTANCE.readJsonToObject( jsonData );
        Map< String, Object > data = ( Map< String, Object > ) map.get( "data" );
        Activity activity = new Activity();
        Channel channel = new Channel();
        Product product = new Product();
        if ( StringUtils.isEmpty( map.get( "activityCode" ) ) ) {
            activity.setActivity_code( map.get( "activityCode" ).toString() );
        }
        if ( StringUtils.isEmpty( map.get( "customerGroupId" ) ) ) {
            activity.setCust_group_id( map.get( "customerGroupId" ).toString() );
            product.setCust_group_id( map.get( "customerGroupId" ).toString() );
            channel.setCust_group_id( map.get( "customerGroupId" ).toString() );
        }
        if ( StringUtils.isEmpty( map.get( "phoneNumber" ) ) ) {
            activity.setAsset_integ_id( map.get( "phoneNumber" ).toString() );
        }
        if ( StringUtils.isEmpty( map.get( "work" ) ) ) {
            activity.setRecommend_description( map.get( "work" ).toString() ); // 需要确认
        }
        if ( StringUtils.isEmpty( map.get( "sms" ) ) ) {
            activity.setChannel_description( map.get( "sms" ).toString() );
        }
        if ( StringUtils.isEmpty( map.get( "activityType" ) ) ) {
            activity.setActivity_type( Integer.parseInt( map.get( "activityType" ).toString() ) );
        }
        if ( StringUtils.isEmpty( map.get( "latn_id" ) ) ) {
            activity.setLatn_id( Integer.parseInt( data.get( "latn_id" ).toString() ) );
        }
        if ( StringUtils.isEmpty( map.get( "productId" ) ) ) {
            product.setProduct_id( map.get( "productId" ).toString() );
            String productName = service.getProductName( map.get( "productId" ).toString() );
            product.setProduct_name( productName );
        }
        if ( StringUtils.isEmpty( map.get( "channelCode" ) ) ) {
            channel.setChannel_id( map.get( "channelCode" ).toString() );
            String contactorName = service.getContactorName( map.get( "channelCode" ).toString() );
            channel.setChannel_name( contactorName );
        }

        service.insertActivity( activity );
        service.insertProduct( product );
        service.insertChannel( channel );

    }*/





}

